/*
 * RReadWriteImplementation.cpp
 *
 *  Created on: 16.08.2011
 *      Author: kirill
 *      Copyright: LGPL
 */

#include "RReadWriteImplementation.h"
#include "Constants.h"

RReadWriteImplementation::RReadWriteImplementation()
{
	// TODO Заполнить этот конструктор тоже
}

RReadWriteImplementation::RReadWriteImplementation( UINT ConnectionId,
													boost::function<bool( const UINT, UINT64 )> ConnectionStatusCallback,
													boost::asio::ip::tcp::endpoint *Endpoint,
													boost::asio::io_service        *Io_service,
													TThreadSafeQueue<STANDART_NETWORK_MESSAGE*> *IncomeQueue):
	connectionId( ConnectionId ), connectionErrorCallback( ConnectionStatusCallback ),
	endpoint( Endpoint ), io_service(Io_service), ConnectionEstabilished( false ), incomeQueue( IncomeQueue ),
    ReadWriteAllow( false), DataArray( NULL ), IncomeMessage( NULL )
{
	outputQueue = new TThreadSafeQueue<STANDART_NETWORK_MESSAGE*>( NETWORK_MIN_QUEUE_LENGHT );
}

RReadWriteImplementation::~RReadWriteImplementation()
{
	delete outputQueue;
}

void
RReadWriteImplementation::ConnectionHandler(const boost::system::error_code &ErrorCode)
{
	fprintf( stderr, "[CONNECTION] ConnectionHandler() %p\n", outputQueue  );
	enm_Errors Error = (ErrorCode) ? ERROR_NETWORK_UNKNOWN_ERROR : ERROR_NETWORK_NONE;
	if ( ( connectionErrorCallback( connectionId, Error ) ) && ( ERROR_NETWORK_NONE == Error ) )
	{
		ConnectionEstabilished = true;
		ReadWriteAllow = true;
		ReadHeader( boost::system::error_code() );
		syncWritingThread = boost::thread( boost::bind( &RReadWriteImplementation::WritingThread, this ) );
	}
	fprintf( stderr, "[CONNECTION] Exit ConnectionHandler() %p\n", outputQueue );
}

void
RReadWriteImplementation::ReadData( const boost::system::error_code &ec )
{
	if ( ( ReadWriteAllow ) && (!ec) )
	{
		fprintf( stderr, "[CONNECTION] ReadData() %p\n", outputQueue );
		if ( ( NULL == DataArray ) && ( IncomeMessage->header.size > 0 ) )
		{
			fprintf( stderr, "[CONNECTION] ReadData() Start reading %u bytes \n", IncomeMessage->header.size );
			DataArrayReadSize = 0;
			DataArray = new BYTE[ IncomeMessage->header.size ];
			IncomeMessage->data = DataArray; // Connect Data to header
			WritingCursor = IncomeMessage->data;

			while ( DataArrayReadSize < IncomeMessage->header.size  ) //Check read message size
			{
				size_t ReadBytes = connectionSocket->read_some( boost::asio::buffer( WritingCursor,
						                                                             IncomeMessage->header.size - DataArrayReadSize ) );
				WritingCursor += ReadBytes; // Go on reading
				DataArrayReadSize += ReadBytes;
				fprintf( stderr, "[CONNECTION] ReadData() Going read %u bytes, stayed - %u \n", ReadBytes, DataArrayReadSize );
			}
		}

		if ( incomeQueue->Send( IncomeMessage ) )
		{
			IncomeMessage = NULL;
			ReadHeader( boost::system::error_code() ); // Run reading header
			fprintf( stderr, "[CONNECTION] ReadData() Read next header %p\n", outputQueue  );
		}
	}
}

void
RReadWriteImplementation::ReadHeader( const boost::system::error_code &ec )
{
	if ( ( ReadWriteAllow ) && (!ec) )
	{
		fprintf( stderr, "[CONNECTION] ReadHeader() %p\n", outputQueue  );
		DataArrayReadSize = 0;
		DataArray = NULL;
		IncomeMessage = new STANDART_NETWORK_MESSAGE();
		connectionSocket->async_receive(  boost::asio::buffer( &IncomeMessage->header, sizeof( NETWORK_MESSAGE_HEADER ) ),
					                      boost::bind(&RReadWriteImplementation::ReadData,
													  this, boost::asio::placeholders::error ) );
	}
}

void
RReadWriteImplementation::WritingThread()
{
	fprintf( stderr, "[CONNECTION] Start writingThread() %p\n", outputQueue  );
	while ( ReadWriteAllow )
	{
		STANDART_NETWORK_MESSAGE *NetworkMessage;
		if ( outputQueue->Receive( &NetworkMessage ) )
		{
			fprintf( stderr, "[CONNECTION] writingThread() %p - Send Message %p\n", outputQueue,  NetworkMessage );
			connectionSocket->send( boost::asio::buffer( NetworkMessage, sizeof( NETWORK_MESSAGE_HEADER ) ));
			size_t SizeToSend = NetworkMessage->header.size; //Calculate size of data block
			if ( ( 0 != SizeToSend ) && ( NULL != NetworkMessage->data ) )
			{
				fprintf( stderr, "[CONNECTION] writingThread() Start DATA, Size: %u\n", SizeToSend );
				BYTE* WritingCursor = NetworkMessage->data;
				while( SizeToSend > 0 )
				{
					size_t SentBytes = connectionSocket->send( boost::asio::buffer( WritingCursor, SizeToSend ));
					WritingCursor += SentBytes; // increase curson on written byte count
					SizeToSend -= SentBytes;
					fprintf( stderr, "[CONNECTION] writingThread() Send %u bytes, stayed - %u\n", SentBytes, SizeToSend );
				}
			}
			delete NetworkMessage;// It deletes its data itself
		}
		else
		{
			boost::this_thread::sleep( boost::posix_time::milliseconds( DEFAULT_NETWORK_WAIT_FOR ) );
		}
	}
	fprintf( stderr, "[CONNECTION] Exit writingThread() %p\n", outputQueue  );
}
